﻿using MySqlSync.Core.Interfaces;
using MySqlSync.Core.Models;

namespace MySqlSync.SyncService
{
    /// <summary>
    /// 同步生产者服务
    /// </summary>
    public class SyncProducerService : ISyncProducerService
    {
        private readonly IDbService _dbService;
        private readonly IKafkaProducerService _kafkaProducer;
        private readonly IRedisService _redisService;
        private readonly SyncConfig _config;
        private int _retryCount = 3;
        private int _retryDelayMs = 2000;

        /// <summary>
        /// 构造函数
        /// </summary>
        public SyncProducerService(
            IDbService dbService,
            IKafkaProducerService kafkaProducer,
            IRedisService redisService,
            SyncConfig config)
        {
            _dbService = dbService ?? throw new ArgumentNullException(nameof(dbService));
            _kafkaProducer = kafkaProducer ?? throw new ArgumentNullException(nameof(kafkaProducer));
            _redisService = redisService ?? throw new ArgumentNullException(nameof(redisService));

            _config = config ?? throw new ArgumentNullException(nameof(config));

            if (string.IsNullOrEmpty(_config.SourceConnectionString))
                throw new ArgumentException("源数据库连接字符串不能为空", nameof(config));

            if (_config.Tables == null || !_config.Tables.Any())
                throw new ArgumentException("至少需要配置一个同步表", nameof(config));
        }

        /// <summary>
        /// 启动生产者服务
        /// </summary>
        public async Task StartAsync(CancellationToken cancellationToken)
        {
            Console.WriteLine("同步生产者服务已启动");
            Console.WriteLine($"需要同步的表: {string.Join(", ", _config.Tables)}");

            // 验证所有表是否存在
            foreach (var tableName in _config.Tables)
            {
                if (!await _dbService.TableExistsAsync(_config.SourceConnectionString, tableName))
                {
                    Console.WriteLine($"警告: 源数据库中不存在表 {tableName}，将跳过该表");
                }
            }

            // 开始同步循环
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    foreach (var tableName in _config.Tables)
                    {
                        if (cancellationToken.IsCancellationRequested)
                            break;

                        // 检查表格是否存在
                        if (!await _dbService.TableExistsAsync(_config.SourceConnectionString, tableName))
                            continue;

                        Console.WriteLine($"开始处理表: {tableName}");

                        // 检查是否需要执行全量同步
                        var isFullSyncCompleted = await _redisService.IsFullSyncCompletedAsync(tableName);

                        if (!isFullSyncCompleted)
                        {
                            // 执行全量同步
                            await PerformFullSyncAsync(tableName, cancellationToken);
                            await _redisService.SetFullSyncCompletedAsync(tableName, true);
                            Console.WriteLine($"表 {tableName} 全量同步完成");
                        }
                        else
                        {
                            // 执行增量同步
                            await PerformIncrementalSyncAsync(tableName, cancellationToken);
                            Console.WriteLine($"表 {tableName} 增量同步完成");
                        }
                    }

                    // 等待下一轮同步
                    Console.WriteLine($"等待 {_config.SyncIntervalSeconds} 秒后进行下一轮同步");
                    await Task.Delay(TimeSpan.FromSeconds(_config.SyncIntervalSeconds), cancellationToken);
                }
                catch (OperationCanceledException)
                {
                    // 预期的取消操作
                    break;
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"同步过程中发生错误: {ex.Message}");
                    Console.WriteLine("将在3秒后重试...");
                    await Task.Delay(3000, cancellationToken);
                }
            }

            // 确保所有消息都已发送
            await _kafkaProducer.FlushAsync();
            Console.WriteLine("同步生产者服务已停止");
        }

        /// <summary>
        /// 执行全量同步
        /// </summary>
        private async Task PerformFullSyncAsync(string tableName, CancellationToken cancellationToken)
        {
            long lastId = await _redisService.GetLastSyncIdAsync(tableName);
            int batchCount = 0;
            int totalRecords = 0;

            try
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    // 获取一批数据
                    var records = await RetryAsync(
                        () => _dbService.GetAllRecordsAsync(
                            _config.SourceConnectionString,
                            tableName,
                            _config.BatchSize,
                            lastId
                        ),
                        cancellationToken
                    );

                    var recordList = records.ToList();
                    if (!recordList.Any())
                    {
                        // 没有更多数据，全量同步完成
                        break;
                    }

                    // 发送到Kafka
                    foreach (var record in recordList)
                    {
                        if (cancellationToken.IsCancellationRequested)
                            break;

                        await RetryAsync<bool>(async () =>
                        {
                            await _kafkaProducer.SendMessageAsync(record);
                            return true; // 实际返回值根据操作逻辑确定
                        }, cancellationToken);


                        // 更新最后同步的ID
                        if (long.TryParse(record.RecordId, out long currentId) && currentId > lastId)
                        {
                            lastId = currentId;
                        }

                        totalRecords++;
                    }

                    batchCount++;
                    Console.WriteLine($"表 {tableName} 全量同步 - 已处理 {totalRecords} 条记录 ({batchCount} 批)");

                    // 保存进度
                    await _redisService.SetLastSyncIdAsync(tableName, lastId);

                    // 短暂延迟，避免过度占用资源
                    if (recordList.Count == _config.BatchSize)
                    {
                        await Task.Delay(100, cancellationToken);
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine($"表 {tableName} 全量同步失败: {ex.Message}");
                // 保存当前进度，以便下次可以从断点继续
                await _redisService.SetLastSyncIdAsync(tableName, lastId);
                throw;
            }
        }

        /// <summary>
        /// 执行增量同步
        /// </summary>
        private async Task PerformIncrementalSyncAsync(string tableName, CancellationToken cancellationToken)
        {
            var lastSyncTime = await _redisService.GetLastSyncTimeAsync(tableName);
            var currentSyncTime = DateTime.Now;

            try
            {
                // 获取增量数据
                var changedRecords = await RetryAsync(
                    () => _dbService.GetChangedRecordsAsync(
                        _config.SourceConnectionString,
                        tableName,
                        lastSyncTime
                    ),
                    cancellationToken
                );

                var recordList = changedRecords.ToList();
                if (!recordList.Any())
                {
                    Console.WriteLine($"表 {tableName} 没有增量数据需要同步");
                    await _redisService.SetLastSyncTimeAsync(tableName, currentSyncTime);
                    return;
                }

                // 按操作时间排序，确保顺序正确
                var orderedRecords = recordList.OrderBy(r => r.OperationTime).ToList();

                // 发送到Kafka
                foreach (var record in orderedRecords)
                {
                    if (cancellationToken.IsCancellationRequested)
                        break;

                    await RetryAsync<bool>(async () =>
                    {
                        await _kafkaProducer.SendMessageAsync(record);
                        return true; // 实际返回值根据操作逻辑确定
                    }, cancellationToken);
                }

                Console.WriteLine($"表 {tableName} 增量同步 - 已处理 {orderedRecords.Count} 条记录");

                // 更新最后同步时间
                await _redisService.SetLastSyncTimeAsync(tableName, currentSyncTime);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"表 {tableName} 增量同步失败: {ex.Message}");
                // 不更新最后同步时间，以便下次重试
                throw;
            }
        }

        /// <summary>
        /// 带重试的异步操作
        /// </summary>
        private async Task<T> RetryAsync<T>(Func<Task<T>> operation, CancellationToken cancellationToken, int? retryCount = null)
        {
            int attempts = 0;
            int maxAttempts = retryCount ?? _retryCount;

            while (true)
            {
                attempts++;
                try
                {
                    return await operation();
                }
                catch (Exception ex)
                {
                    if (attempts >= maxAttempts)
                    {
                        Console.WriteLine($"达到最大重试次数 ({maxAttempts})，操作失败: {ex.Message}");
                        throw;
                    }

                    Console.WriteLine($"操作失败，将在 {_retryDelayMs}ms 后重试 (第 {attempts}/{maxAttempts} 次): {ex.Message}");
                    await Task.Delay(_retryDelayMs, cancellationToken);
                }
            }
        }
    }

}
